package com.hqq.concurrency.reactor.framework;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;

/**
 * NIODatagramChannel
 * <p>
 * Created by heqianqian on 2017/8/5.
 */
public class NIODatagramChannel extends AbstractNIOChannel {

    private static final Logger LOGGER = LoggerFactory.getLogger(NIODatagramChannel.class);

    private int port;

    /**
     * 创建一个新的通道
     *
     * @param port    端口
     * @param handler 处理器
     */
    public NIODatagramChannel(int port, ChannelHandler handler) throws IOException {
        super(DatagramChannel.open(), handler);
        this.port = port;
    }

    @Override
    public int getInterestedOps() {
        //无需连接 直接读取
        return SelectionKey.OP_READ;
    }

    @Override
    public void bind() throws IOException {
        ((ServerSocketChannel) getJavaChannel()).socket().bind(new InetSocketAddress(InetAddress.getLocalHost(), port));
        getJavaChannel().configureBlocking(false);
        LOGGER.info("Bind UDP Socket at port {}", port);
    }

    @Override
    public DatagramPacket read(SelectionKey key) throws IOException {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        SocketAddress sender = ((DatagramChannel) key.channel()).receive(byteBuffer);

        byteBuffer.flip();
        DatagramPacket datagramPacket = new DatagramPacket(byteBuffer);
        datagramPacket.setSender(sender);

        return datagramPacket;
    }

    @Override
    protected void doWrite(Object pendingWrite, SelectionKey key) throws IOException {
        DatagramPacket pendingPacket = (DatagramPacket) pendingWrite;
        ((DatagramChannel) getJavaChannel()).send(pendingPacket.getData(), pendingPacket.getReceiver());
    }


    public static class DatagramPacket {
        private SocketAddress sender;
        private ByteBuffer data;
        private SocketAddress receiver;

        public DatagramPacket(ByteBuffer data) {
            this.data = data;
        }

        public SocketAddress getSender() {
            return sender;
        }

        public void setSender(SocketAddress sender) {
            this.sender = sender;
        }

        public SocketAddress getReceiver() {
            return receiver;
        }

        public void setReceiver(SocketAddress receiver) {
            this.receiver = receiver;
        }

        public ByteBuffer getData() {
            return data;
        }
    }
}
